package com.atguigu.userprofile.app

import java.util.Properties

import com.atguigu.userprofile.bean.TagInfo
import com.atguigu.userprofile.dao.TagInfoDAO
import com.atguigu.userprofile.util.MyPropertiesUtil
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

/**
 * 将每天计算的多个标签任务表,进行合并,形成宽表,并存入到ClickHouse中.
 * 格式: uid,性别,几零后,最近连续30天下单数 .....
 * 用户标识,多个标签任务计算结果
 */
object TagMergeApp {


  /*
  1、要组合哪些标签表？  show tables (不准确)
    最好 查询 tag_info  join  task_info
    获得启用状态的标签列表
  2、建立宽表
    表名     user_tag_merge_20210609  ......
    字段     uid ,  <tag_code1> ,<tag_code2>
    考虑到每天的标签数是不一样的，那么宽表的字段也不一样，所以不使用一张固定的表，而是每天建一张新表，每天的字段可以不同。
  3、合并数据进宽表 : 利用spark-sql中 pivot 进行行转列操作。
   */
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setAppName("task_tag_merge_app")//.setMaster("local[*]")
    val sparkSession: SparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()

    // 在最终提交给spark-submit参数后,需要再传如下两个参数: 参数1-任务id,参数2-执行时间 yyyyMMdd格式 10 2021-06-08
    val taskId: String = args(0)
    val taskDate: String = args(1)

    val properties: Properties = MyPropertiesUtil.load("config.properties")
    val hdfsPath: String = properties.getProperty("hdfs-store.path")
    val dwDbName: String = properties.getProperty("data-warehouse.dbname")
    val upDbName: String = properties.getProperty("user-profile.dbname")

    /*
      1、查询 tag_info  join  task_info
      获得启用状态的标签列表,也就是获取三级标签的名称作为建表的列.
      select tg.* from tag_info tg join task_info tk on tg.tag_task_id = tk.id where  tk.task_status='1';
      List(
        TagInfo(13,Tag_Natural_Characteristic_Gender,用户性别,3,5,1,3,null,null,1,null,2022-01-19 10:00:41.0,null),
        TagInfo(17,Tag_Natural_Characteristic_AgeGroup,年龄段统计,3,5,1,3,null,null,2,null,2022-01-20 15:05:13.0,null),
        TagInfo(24,Tag_Order_Habit_Order_Last_30d_Count,最近30天下单数量,3,9,1,1,null,null,3,null,2022-01-20 16:39:26.0,null))
     */
    val tagInfoList: List[TagInfo] = TagInfoDAO.getTagInfoListOnTask()


    /*
      2、建立宽表,这里定义的表名为 user_tag_merge_yyyyMMdd,因为每条的标签个数不同,可能每天都会添加新的任务,这里不采用时间按照提天进行分区,而是直接创建当前的表.
      user_tag_merge_$taskDate  ......
      */
    val tableName = s"user_tag_merge_${taskDate.replace("-", "")}"

    /** user_tag_merge_20210608 */

    /** tag_natural_characteristic_gender string ,tag_natural_characteristic_agegroup string ,tag_order_habit_order_last_30d_count string */
    val tagColNames = tagInfoList.map(_.tagCode.toLowerCase + " string ").mkString(",")

    val dropTableSql = s" drop table if exists $tableName"

    /*
      -- 每天标签任务执行结果,组合宽表 建表语句.
      -- 其中 表名每天会变化: user_tag_merge_20210608,字段也会根据当前计算的三级标签动态生成.
      create table user_tag_merge_20210608
      (
          uid                                  string,
          tag_natural_characteristic_gender    string, -- 三级标签的 tag_code转化为小写.
          tag_natural_characteristic_agegroup  string,
          tag_order_habit_order_last_30d_count string
      )
          comment '标签宽表'
          ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
          LOCATION 'hdfs://hadoop102:8020/user_profile/user_profile0111/user_tag_merge_20210608';
           */
    val createTableSql =
      s"""
         |create table  $tableName ( uid string ,$tagColNames )
         |     comment '标签宽表'
         |     ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t'
         |     LOCATION    '$hdfsPath/$upDbName/$tableName'
       """.stripMargin

    /*
    3、往宽表中灌数,并将长表转化为宽表.
      使用 pivot进行行转列操作.这里要区分开三种类型的字段: 维度字段,参与group by的字段. 聚合字段 sum(if(条件,满足的值,null)) . 旋转的字段.
      如果确定好了三种字段,则使用固定的语法格式,开始进行行专列操作. 具体的测试过程,可以看pivot_测试类中.
      另外,在我们使用固定的语法时,仅仅需要我们指定:聚合列和旋转列,如果其他列不指定,则都是维度列.
      TODO 开始组合我们的标签宽表。
       1、先将每天所有的标签任务表名获取到，使用 union all 构建标签任务长表。具体的字段为 uid，tag_code,tag_value.
       2、使用`pivot`将长表转化为宽表
       3、往创建好的宽表中灌数据. ==> 当天所有的标签任务表中数据,插入到
     */

    /*
        -- 每天的标签任务表转化成长表,使用union all,这里需要注意:在使用 union all时,多个表中的参数列需要对齐,这里的对齐指的是: 列个数相同,列的类型相同.
        select uid, tag_value from tag_natural_characteristic_gender -- 三级标签的表名
        union all
        select uid, tag_value from tag_natural_characteristic_agegroup
        union all
        select uid, cast(tag_value as string) from tag_order_habit_order_last_30d_count
        ;
     */
    val unionSql = tagInfoList.map(tagInfo => {
      s"select uid ,'${tagInfo.tagCode.toLowerCase}' as  tag_code,tag_value from ${tagInfo.tagCode.toLowerCase} "
    }).mkString(" union all ")
    /** 拼接旋转列的值,具体用在使用pivot中 for 旋转列 in('旋转列值1','旋转列值2','旋转列值3','旋转列值4' .... ) */
    val inSql: String = tagInfoList.map("'" + _.tagCode.toLowerCase + "'").mkString(",")
    /*
      拼接查询语句,这里需要具体的行转列过程,也就是pivot, 旋转列:每个用户的多个标签任务的列,也就是(三级标签的名称 tag_name). 聚合列:tag_value,这里每个用户对应一个标签任务值.虽然不涉及聚合操作,但是也要写.这里直接使用拼接代替.
     */
    val selectSql =
      s""" select * from   (  $unionSql )
         |    pivot ( concat_ws(',' ,collect_set(tag_value)) as tv
         |     for  tag_code in( $inSql))
         |
       """.stripMargin
    /*
        TODO 最终的拼接SQL的样式如下
          insert overwrite table user_tag_merge_20210608
          select *
          from (
               select uid, 'tag_natural_characteristic_gender' as tag_code, tag_value
               from tag_natural_characteristic_gender
               union all
               select uid, 'tag_natural_characteristic_agegroup' as tag_code, tag_value
               from tag_natural_characteristic_agegroup
               union all
               select uid, 'tag_order_habit_order_last_30d_count' as tag_code, tag_value
               from tag_order_habit_order_last_30d_count
              )
              pivot ( concat_ws(',' ,collect_set(tag_value)) as tv
              for  tag_code in( 'tag_natural_characteristic_gender','tag_natural_characteristic_agegroup','tag_order_habit_order_last_30d_count'))
     */
    val insertSql = s"insert overwrite table $tableName  $selectSql"

    sparkSession.sql(s"use $upDbName")
    sparkSession.sql(dropTableSql)
    sparkSession.sql(createTableSql)
    sparkSession.sql(insertSql)
  }
}
